package cc.rboot.topic.processor;

import cc.rboot.topic.core.AbstractMessageProcessor;
import cc.rboot.topic.core.IMessageConsumer;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

public class LocalQueueMessageProcessor<T> extends AbstractMessageProcessor<T> {

    private final Map<IMessageConsumer<T>, ConcurrentLinkedQueue<T>> map = new ConcurrentHashMap<>();

    @Override
    public void register(IMessageConsumer<T> consumer) {
        ConcurrentLinkedQueue<T> queue = new ConcurrentLinkedQueue<>();
        map.put(consumer, queue);

        new Thread(() -> {
            while (true) {
                try {
                    T message = this.poll(consumer);
                    if (message == null) {
                        continue;
                    }
                    boolean ifSuccess = false;
                    try {
                        ifSuccess = consumer.onMessage(message);
                    } catch (Exception ex) {
                        // Do nothing
                    }
                    if (!ifSuccess) {
                        try {
                            this.sendToConsumer(queue, message);
                        } catch (Exception ex) {
                            // 打印下日志即可，一般消费者都是回调，出错了也没有报错的对象
                            System.out.println(ex.getMessage());
                        }
                    }
                    continue;
                } catch (Exception ex) {
                    System.out.println("发生错误" + ex.getMessage());
                }
                this.sleep();
            }
        }).start();
    }


    private T poll(IMessageConsumer<T> consumer) throws Exception {
        return map.get(consumer).poll();
    }

    @Override
    public void send(T message) throws Exception {
        map.forEach((consumer, queue) -> {
            try {
                this.sendToConsumer(queue, message);
            } catch (Exception exception) {
                exception.printStackTrace();
            }
        });
    }

    public void sendToConsumer(ConcurrentLinkedQueue<T> queue, T message) throws Exception {
        queue.add(message);
    }

    /**
     * 单轮消费间隔时间
     */
    private void sleep() {
        try {
            Thread.sleep(20);
        } catch (InterruptedException e) {
            // Do nothing
        }
    }
}
